package com.navinfo.tripanalysis.offline;

import com.navinfo.location.mileage.bean.MileageAndFuel;
import com.navinfo.platform.score.TripScoreService;
import com.navinfo.platform.score.pojo.ScoreWeight;
import com.navinfo.tripanalysis.common.arithmetic.common.DrivingData;
import com.navinfo.tripanalysis.common.arithmetic.common.EngineAggregatedData;
import com.navinfo.tripanalysis.common.arithmetic.common.OuterEventData;
import com.navinfo.tripanalysis.common.arithmetic.common.OuterStatisticData;
import com.navinfo.tripanalysis.common.config.LoadConf;
import com.navinfo.tripanalysis.common.pojo.Point;
import com.navinfo.tripanalysis.common.pojo.Point0200;
import com.navinfo.tripanalysis.common.pojo.RunningTypeEnum;
import com.navinfo.tripanalysis.common.util.ArithmeticUtils;
import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.common.util.DateUtils;
import com.navinfo.tripanalysis.common.vehicle.VehicleInfo;
import com.navinfo.tripanalysis.offline.pojo.AlgorithmChainCommand;
import com.navinfo.tripanalysis.offline.pojo.AlgorithmOuterData;
import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import com.navinfo.tripanalysis.offline.pojo.PreTripInfo;
import com.navinfo.tripanalysis.offline.service.*;
import com.navinfo.tripanalysis.offline.service.impl.ConfigFileServiceYamlImpl;
import com.navinfo.tripanalysis.offline.util.TripStatisticFixUtils;
import com.navinfo.tripanalysis.offline.util.TripStatisticUtils;
import org.apache.commons.cli.*;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.storage.StorageLevel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.*;
import java.util.stream.Collectors;

import static com.navinfo.tripanalysis.offline.util.SparkAppUtils.getSparkBuilder;

/**
 * 驾驶行为分析离线模块<br/>
 * 通过0200、0f37点来计算，并且落盘公共协议数据
 * @author 沈东生、王恩宝
 */
public class TripAnalysisApplication {
    private static final Logger logger = LoggerFactory.getLogger(TripAnalysisApplication.class);

    /**
     * 系统运行类型，1东风，2青汽
     */
    static RunningTypeEnum runningTypeEnum;
    /**
     * 开始时间毫秒数
     */
    static Long start;
    /**
     * 结束时间毫秒数
     * */
    static Long end;
    /**
     * 一天的毫秒数
     */
    static final Long DAY_MS = 86400000L;

    public static void main(String[] args) throws Exception {
        long startTime = System.currentTimeMillis();
        logger.error("开始离线驾驶行为分析，args:{}", Arrays.toString(args));

        //加载配置
        Options options = setCmdOptions();
        CommandLine cmd = parseCmdOptions(options, args);
        Properties config = loadProps(cmd);
        parseDays(args, cmd);

        String runType = config.getProperty("system.running.type");
        if(StringUtils.isEmpty(runType)){
            logger.error("请配置系统运行类型，1东风，2青汽");
            System.exit(1);
        }

        runningTypeEnum = RunningTypeEnum.valueOf(Integer.parseInt(runType));
        logger.error("离线驾驶行为分析，系统运行类型：{},{}", runningTypeEnum.getType(), runningTypeEnum.getDesc());


        //加载回放特定tid的列表
        LoadDataParam param = new LoadDataParam();
        param.setTidList(loadTidList(cmd));

        //初始化SparkSession
        SparkSession spark = getSparkBuilder(config).getOrCreate();
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());

        //创建相关服务
        ServiceFactory factory = new ServiceFactory(config, runningTypeEnum);
        DataBatchService dataBatchService = factory.createDataBatchService();
        LoadPoint0F37Service load0F37Service = factory.createLoad0F37DataService();
        LoadPoint0200Service load0200Service = factory.createLoad0200DataService();
        PointSortGroupService sortGroupService = factory.createPointSortGroupService();
        DrivingDataService drivingDataService = factory.createDrivingDataService();
        List<SaveTripStatisticService> saveTripStatisticServices = factory.createSaveTripServices();
        List<SaveTripEventService> saveTripEventServices = factory.createSaveEventServices();
        SavePointProtocolService savePointProtocolService = factory.createSavePointService();
        TripStatisticExtendService tripStatisticExtendService = factory.createTripStatisticExtendService();

        //是否加载瓦片数据
        boolean loadTileData = Boolean.parseBoolean(config.getProperty("load.data.tile.open"));
        logger.error("loadTileData:{}", loadTileData);

        //广播变量：瓦片与区域信息、车辆配置信息、算法配置信息
        final Broadcast<Map<Long, Integer>> tileLcBroadcast = jsc.broadcast(loadTileData ? new HashMap<>(factory.createLoadTileService().load(spark, jsc).collectAsMap()) : new HashMap<>() );
        final Broadcast<Map<Long, VehicleInfo>> vehicleInfoBroadcast = vehicleInfoBroadcast(spark, jsc, factory );
        final Broadcast<Map> algorithmConfBroadcast = algorithmConfBroadcast(jsc, config);

        //添加行程打分算法与权重信息
        Map<String, ScoreWeight> score = new HashMap<>(TripScoreService.getLoadAndEnvironmentData());
        logger.error("scoreSize:{},score1Size:{}",score.size());
        final Broadcast<Map<String, ScoreWeight>> scoreBroadcast = jsc.broadcast(score);

        //广播变量：系统运行类型
        final Broadcast<RunningTypeEnum> runningTypeBroadcast = jsc.broadcast(runningTypeEnum);

        //按天进行行程和事件的计算
        for(long sTime=start; sTime<end; sTime+= DAY_MS){
            Date currentDay = new Date(sTime);
            param.setDay(currentDay);

            //获取昨天的中间变量
            final Broadcast<Map<Long, DrivingData>> yesterdayDrivingData = drivingDataBroadcast(spark, jsc, drivingDataService, sTime);

            //最终的输出结果
            List<JavaPairRDD<Long, AlgorithmOuterData>> outDataRDDs = new ArrayList<>();

            //按批次处理数据
            List<LoadDataParam> batchParams = dataBatchService.createBatch(param);
            for (LoadDataParam oneBatchParam : batchParams) {
                logger.error("========>oneBatchParam:{}", oneBatchParam);

                //加载0200、0F37的数据
                JavaRDD<Point> points0F37 = load0F37Service.load(spark, jsc, oneBatchParam);
                JavaRDD<Point> points0200 = load0200Service.load(spark, jsc, oneBatchParam);
                if (points0F37 == null) {
                    logger.error("OOO:0f37 data is null");
                    points0F37 = jsc.parallelize(new ArrayList<>());
                }
                if (points0200 == null) {
                    logger.error("OOO:0200 data is null");
                    points0200 = jsc.parallelize(new ArrayList<>());
                }

                logger.error("OOO:0f37 data size:{}", points0F37.count());
                logger.error("OOO:0200 data size:{}", points0200.count());

                //分组排序0200、0F37数据
                JavaPairRDD<Long, List<Point>> tidPoints0F37 = sortGroupService.sortAndGroup(points0F37);
                JavaPairRDD<Long, List<Point>> tidPoints0200 = sortGroupService.sortAndGroup(points0200);

                //将0F37和0200数据连接
                JavaPairRDD<Long, AlgorithmOuterData> outDataRDD = tidPoints0F37.fullOuterJoin(tidPoints0200)
                    .mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, Tuple2<Optional<List<Point>>, Optional<List<Point>>>>>, Long, AlgorithmOuterData>) it -> {
                        return new Iterator<Tuple2<Long, AlgorithmOuterData>>() {
                            private final Map algorithmConf = algorithmConfBroadcast.getValue();
                            final AlgorithmChainService algorithmChain = factory.createAlgorithmChainService(runningTypeBroadcast.getValue(),algorithmConf);
                            final PreTripInfoService preTripService = factory.createServiceImpl(runningTypeBroadcast.getValue(), PreTripInfoService.class);

                            @Override
                            public boolean hasNext() {
                                return it.hasNext();
                            }

                            @Override
                            public Tuple2<Long, AlgorithmOuterData> next() {
                                Tuple2<Long, Tuple2<Optional<List<Point>>, Optional<List<Point>>>> pointTuple2 = it.next();
                                Long tid = pointTuple2._1();
                                List<Point> points0F37 = null;
                                List<Point0200> points0200 = null;

                                if (pointTuple2._2._1.isPresent()) {
                                    points0F37 = pointTuple2._2._1.get();
                                }
                                if (pointTuple2._2._2.isPresent()) {
                                    points0200 = (List) pointTuple2._2._2.get();
                                }

                                //如果只根据0f37来生成行程开关，当前没有0f37点，则直接返回一个空对像
                                if (Boolean.valueOf(config.getProperty("only0F37Trip")) && CommonUtils.isCollectionEmpty(points0F37)) {
                                    return new Tuple2<>(tid, new AlgorithmOuterData());
                                }

                                //构建算法链的输出对象
                                AlgorithmOuterData outerData = new AlgorithmOuterData();
                                outerData.setOutStatistics(new ArrayList<>());
                                outerData.setOutEvents(CommonUtils.isCollectionNotEmpty(saveTripEventServices) ? new ArrayList<>() : null);

                                //如果设置了保存公共轨迹协议服务
                                if (null != savePointProtocolService) {
                                    outerData.setPoints0200(points0200);
                                    outerData.setPoints0f37(points0F37);
                                }

                                //初始化中间数据
                                DrivingData drivingData = yesterdayDrivingData.getValue().get(tid);
                                if (drivingData == null) {
                                    drivingData = new DrivingData();
                                    drivingData.getCommonData().setTid(tid);
                                }

                                drivingData.setVehicleInfo(vehicleInfoBroadcast.getValue().get(tid));
                                drivingData.setTileLcBroadcast(tileLcBroadcast);
                                logger.error("drivingData:{}", drivingData);

                                //算法链执行的命令对象
                                AlgorithmChainCommand command = new AlgorithmChainCommand();
                                command.setTid(tid);
                                command.setDay(currentDay);
                                command.setOuterData(outerData);
                                command.setDrivingData(drivingData);

                               /*行程切分具体流程<br/>
                                1）根据当天0200点，按发动机转速>100切分出子行程，统计出行程开始、结束时间及行程时长；<br/>
                                2）根据子行程的开始结束时间，切分出对应的0f37及0200的子数据；<br/>
                                3) 根据0200的子行程调用精准里程油耗算法算出子行程的精准里程、精准油耗；<br/>
                                4）根据子行程的0200、0F37数据，调用算法链，算出对应的积分里程、积分油耗、积分时长及算法的单项指标值；<br/>
                                5）根据当天的0200点，调用精准里程油耗算法算出当天总的精准里程油耗值；<br/>
                                6）根据子行程的精准里程(油耗)占所有行程的总精准里程(油耗)比例 * 当天总的精准里程(油耗)，算出每个子行程的精准里程（油耗）<br/>
                                7）使用6相同的算法，计算出行程中其他指标的精准值；<br/>
                                */
                                List<PreTripInfo> tripList = preTripService.preJudgeTrip(points0F37, points0200, currentDay);
                                if (CommonUtils.isCollectionNotEmpty(tripList)) {
                                    int tripSize = tripList.size();
                                    if (logger.isInfoEnabled()) {
                                        logger.info("tid:{},day:{},tripSize:{}", tid, currentDay, tripSize);
                                    }

                                    //步骤3切分行程的精准里程和油耗数组
                                    Integer[] tripFuel = new Integer[tripSize];
                                    Integer[] tripMileage = new Integer[tripSize];

                                    for (int i = 0; i < tripSize; i++) {
                                        PreTripInfo preTripInfo = tripList.get(i);
                                        tripFuel[i] = preTripInfo.getPreciseFuel();
                                        tripMileage[i] = preTripInfo.getPreciseMileage();

                                        command.setPoints0f37(preTripInfo.getSubPoints0f37());
                                        command.setPoints0200(preTripInfo.getSubPoints0200());
                                        if (i == (tripSize - 1)) {
                                            command.setHandleCrossDay(true);
                                        } else {
                                            command.setHandleCrossDay(false);
                                        }

                                        if (logger.isInfoEnabled()) {
                                            logger.info("--->i={}", i);
                                            logger.info("----{}\n", preTripInfo);
                                            long lastGpsTime = drivingData.getCommonData().getLastGpsTime();
                                            if (lastGpsTime <= 0) {
                                                logger.info("--->lastGpsTime is null");
                                            } else {
                                                logger.info("--->lastGpsTime: {}", DateUtils.format(new Date(lastGpsTime), DateUtils.DateFormat.YYYY_MM_DD_HH_MM_SS));
                                                if (lastGpsTime > preTripInfo.getStartTime()) {
                                                    logger.error("Error:lastGpsTime>info.getStartTimeMs");
                                                }
                                            }
                                        }

                                        //4、执行算法链，算出行程对应的积分里程、积分油耗、积分时长及算法的单项指标值
                                        algorithmChain.execute(command);
                                    }

                                    //5、根据0200计算出当天总的精准里程油耗值
                                    MileageAndFuel mileageAndFuel = TripStatisticUtils.calcMileageFuel(runningTypeBroadcast.getValue(), points0200);
                                    int dayMileage = ArithmeticUtils.mileageKMToM(mileageAndFuel.getStdMileage());
                                    int dayFuel = ArithmeticUtils.fuelLToMl(mileageAndFuel.getStdFuelCon());

                                    //6、根据子行程的精准里程(油耗)占所有行程的总精准里程(油耗)比例 * 当天总的精准里程(油耗)，算出每个子行程的精准里程（油耗）
                                    List<OuterStatisticData> outTrips = outerData.getOutStatistics();
                                    if (outTrips.size() == tripSize) {
                                        //精准后各个行程的里程油耗
                                        Integer[] correctMileage = TripStatisticFixUtils.fixArrayData(dayMileage, tripMileage);
                                        Integer[] correctFuel = TripStatisticFixUtils.fixArrayData(dayFuel, tripFuel);
                                        for (int i = 0; i < tripSize; i++) {
                                            PreTripInfo preTripInfo = tripList.get(i);
                                            OuterStatisticData statisticData = outTrips.get(i);
                                            statisticData.setTripId(preTripInfo.getTripId());
                                            statisticData.setForceEndFlag(preTripInfo.getForceEndFlag());

                                            //根据0200精准行程的里程油耗
                                            statisticData.setTripFuel(correctFuel[i]);
                                            statisticData.setTripMileage(correctMileage[i]);
                                            statisticData.setPreciseFlag(OuterStatisticData.PRECISE_YES);

                                            //根据0200精准行程开始、结束的相关信息
                                            statisticData.setRouteStartTime(preTripInfo.getStartTime() / 1000L);
                                            statisticData.setRouteStartLatitude(preTripInfo.getStartLatitude());
                                            statisticData.setRouteStartLongitude(preTripInfo.getStartLongitude());
                                            statisticData.setRouteStartHeight(preTripInfo.getStartHeight());
                                            statisticData.setRouteStartDirection(preTripInfo.getStartDirection());
                                            statisticData.setRouteEndTime(preTripInfo.getEndTime() / 1000L);
                                            statisticData.setRouteEndLongitude(preTripInfo.getEndLongitude());
                                            statisticData.setRouteEndLatitude(preTripInfo.getEndLatitude());
                                            statisticData.setRouteEndHeight(preTripInfo.getEndHeight());
                                            statisticData.setRouteEndDirection(preTripInfo.getEndDirection());
                                            statisticData.setTripDuration((int) (statisticData.getRouteEndTime() - statisticData.getRouteStartTime()));
                                            if (tileLcBroadcast != null) {
                                                Map<Long, Integer> value = tileLcBroadcast.getValue();
                                                statisticData.setRouteStartLc(ArithmeticUtils.getLc(statisticData.getRouteStartLatitude(), statisticData.getRouteStartLongitude(), value));
                                                statisticData.setRouteEndLc(ArithmeticUtils.getLc(statisticData.getRouteEndLatitude(), statisticData.getRouteEndLongitude(), value));
                                            }

                                            //7）使用6相同的算法，计算出行程中其他指标的精准值；
                                            TripStatisticFixUtils.fixTrips(statisticData);

                                            //通过0200精准行程开始结束时的里程、油耗
                                            TripStatisticUtils.calcStartEndMileageFuel(statisticData, preTripInfo.getSubPoints0200());

                                            //计算行程打分
                                            TripStatisticUtils.calcTripScore(statisticData, scoreBroadcast);
                                        }
                                    } else {
                                        logger.error("OOO->preTrip tripSize is not equals to trip tripSize tid:{},pre_size:{},trip_size:{}", tid, tripSize, outTrips.size());
                                    }
                                }

                                //清空drivingdata上无用的数据
                                DrivingData outDrivingData = outerData.getDrivingData();
                                if (outDrivingData != null) {
                                    outDrivingData.setVehicleInfo(null);
                                    outDrivingData.setTileLcBroadcast(null);
                                    outDrivingData.setUpdateDay(DateUtils.format(currentDay, DateUtils.DateFormat.YYYYMMDD));
                                }

                                logger.info("tid:{},day:{},tripSize:{},eventSize:{}", tid, currentDay, outerData.getOutStatistics().size(), outerData.getOutEvents().size());
                                return new Tuple2<>(tid, outerData);
                            }
                        };
                    });

                outDataRDDs.add(outDataRDD);
                outDataRDD.persist(StorageLevel.MEMORY_AND_DISK_SER());
                logger.error("========>结束一个批次：oneBatchParam:{}", oneBatchParam);

                //执行算法链
                logger.error("========>当前批次保存的count:" + outDataRDD.count());

                //手动清理下资源
                points0F37.unpersist();
                points0200.unpersist();
                tidPoints0F37.unpersist();
                tidPoints0200.unpersist();
            }

            //合并分批处理的数据结果
            JavaPairRDD<Long, AlgorithmOuterData> outData = null;
            if (outDataRDDs.size() > 0) {
                outData = outDataRDDs.get(0);
                for (int i=1; i<outDataRDDs.size(); i++) {
                    JavaPairRDD<Long, AlgorithmOuterData> tmpRdd = outDataRDDs.get(i);
                    if (tmpRdd != null) {
                        outData = outData.union(tmpRdd);
                    }
                }
            }

            //行程、事件、中间数据落盘
            if (outData != null) {
                if (CommonUtils.isCollectionNotEmpty(saveTripStatisticServices)) {
                    logger.error("行程数据落盘...");
                    saveTripData(spark, jsc, saveTripStatisticServices, tripStatisticExtendService, sTime, outData);
                }

                if ( CommonUtils.isCollectionNotEmpty(saveTripEventServices)) {
                    logger.error("事件数据落盘...");
                    saveEventData(spark, jsc, saveTripEventServices, sTime, outData);
                }
                if (null != drivingDataService) {
                    logger.error("中间数据落盘...");
                    saveDrivingData(spark, jsc, drivingDataService, sTime, outData, yesterdayDrivingData);
                }
                if(null != savePointProtocolService){
                    logger.error("公共轨迹协议落盘...");
                    savePointProtocolService.save(spark, outData, sTime);
                }
            } else {
                logger.error("无行程数据生成!!!");
            }
        }

        logger.error("完成离线驾驶行为分析，耗时：{}ms", System.currentTimeMillis()-startTime);
        jsc.close();
        System.exit(0);
    }

    /**
     * 配置命令行参数
     */
    static Options setCmdOptions() {
        Options options = new Options();
        Option opSt = new Option("st", "startTime", true, "[必填参数]开始日期，格式yyyy-MM-dd（包含）");
        Option opEt = new Option("et", "endTime", true, "结束日期，格式yyyy-MM-dd（不包含）");
        opEt.setRequired(false);
        options.addOption(opSt);
        options.addOption(opEt);
        options.addOption(new Option("tids", "terminalIds", true, "按逗号分隔的terminalId进行回放，如果未设此参数，则为全量数据回放"));
        options.addOption(new Option("tidsFile", "terminalIdsFile", true, "terminalIds的文件路径，一行一个terminalId"));
        options.addOption(new Option("h", "help", false, "打印帮助"));

        OptionBuilder.withArgName("property=value");
        OptionBuilder.hasArgs(2);
        OptionBuilder.withValueSeparator();
        OptionBuilder.withDescription("其它的配置项，可覆盖配置文件中的配置项(property=value)");
        Option property = OptionBuilder.create("D");
        options.addOption(property);
        return options;
    }

    /**
     * 解析CMD参数，args中通过"-K V"配置属性
     */
    static CommandLine parseCmdOptions(Options options, String[] args) throws ParseException {
        CommandLine cmd = new PosixParser().parse(options, args);
        if (cmd.hasOption("h")) {
            printUsage(options);
            System.exit(0);
        } else if (!cmd.hasOption("st")) {
            printUsage(options);
            System.exit(1);
        }

        return cmd;
    }

    /**
     * 打印帮助信息
     */
    static void printUsage(Options options) {
        HelpFormatter formatter = new HelpFormatter();
        formatter.printHelp("Application", options);
        System.out.println();
    }

    /**
     * 加载config.yaml的配置<br/>
     * CMD中的-D配置能覆盖配置文件中的配置
     */
    static Properties loadProps(CommandLine cmd) throws URISyntaxException, FileNotFoundException {
        //jar包的位置
        File directory = new File(TripAnalysisApplication.class.getProtectionDomain().getCodeSource().getLocation().toURI().getPath()).getParentFile();
        Properties config = new ConfigFileServiceYamlImpl().setInputStream(new FileInputStream(directory.getAbsolutePath() + "/config.yaml")).load();

        //命令行中的-D覆盖config.yaml的配置
        java.util.Optional.ofNullable(cmd.getOptionProperties("D")).ifPresent(e ->{
            e.forEach((k, v) -> config.put(k, v));
        });

        logger.error("OOO:props开始:-----------");
        for (String s : config.stringPropertyNames()) {
            System.out.println(s + "->" + config.getProperty(s));
        }
        logger.error("OOO:props结束：--------------");
        return config;
    }

    /**
     * 解析开始和结束时间<br/>
     * 时间格式：YYYY_MM_DD或YYYYMMDD
     * @param args
     * @param cmd
     */
    static void parseDays(String[] args, CommandLine cmd) {
        if (args.length >= 1) {
            start = DateUtils.parseDate(cmd.getOptionValue("st")).getTime();
            String endTimeStr = cmd.getOptionValue("et");
            end = StringUtils.isEmpty(endTimeStr) ? (start + DAY_MS) :DateUtils.parseDate(endTimeStr).getTime();
        } else {
            throw new RuntimeException("时间不能为空！");
        }
    }

    /**
     * 通过命令行或文件加载特定回放的tid列表
     * @param cmd
     * @throws IOException
     */
    static List<Long> loadTidList(CommandLine cmd) throws IOException {
        List<Long> tidList = null;
        if (cmd.hasOption("tids")) {
            String tidStr = cmd.getOptionValue("tids");
            if (StringUtils.isNotEmpty(tidStr)) {
                logger.error("load filter tids from CMD:{}", tidStr);
                String[] ss = tidStr.split(",");
                tidList = new ArrayList<>();
                for (String s : ss) {
                    if (StringUtils.isNotEmpty(s)) {
                        tidList.add(Long.valueOf(s));
                    }
                }
                logger.error("filter tids count:{}", tidList.size());
            }
        } else if (cmd.hasOption("tidsFile")) {
            File tidFile = new File(cmd.getOptionValue("tidsFile"));
            logger.error("load filter tidList from file:{}", tidFile.getAbsolutePath());
            if (null!=tidFile && tidFile.exists() && tidFile.isFile()) {
                tidList = FileUtils.readLines(tidFile, "UTF-8").stream().map(t -> t.trim()).filter(t -> !StringUtils.startsWith(t, "#") && t.length() > 0).map(t -> Long.valueOf(t)).collect(Collectors.toList());
                logger.error("filter tidList count:{}", tidList.size());
            } else {
                logger.error("can't load filter tidList file:{}", tidFile.getAbsolutePath());
                System.exit(1);
            }
        }
        return tidList;
    }

    /**
     * 广播车辆信息
     */
    static Broadcast<Map<Long, VehicleInfo>> vehicleInfoBroadcast(SparkSession spark, JavaSparkContext jsc, ServiceFactory factory) throws InstantiationException, IllegalAccessException {
        return jsc.broadcast(new HashMap<>(factory.createLoadVehicleInfoService().load(spark).collectAsMap()));
    }

    /**
     * 广播算法配置信息
     */
    static Broadcast<Map> algorithmConfBroadcast(JavaSparkContext jsc, Properties props) {
        Map algConfig = new HashMap<>(LoadConf.LoadYaml("config_common.yaml"));
        algConfig.putAll(LoadConf.LoadYaml("config_threshold.yaml"));
        algConfig.put("logger.whitelist.terminalIds", Optional.ofNullable(props.getProperty("logger.whitelist.terminalIds")).orElse(""));
        return jsc.broadcast(algConfig);
    }

    /**
     * 广播昨天的中间计算结果
     */
    static Broadcast<Map<Long, DrivingData>> drivingDataBroadcast(SparkSession spark, JavaSparkContext jsc, DrivingDataService drivingDataService, long day) {
        return jsc.broadcast(drivingDataService !=null ? new HashMap<>(drivingDataService.load(spark, jsc, day).collectAsMap()) : new HashMap<>());
    }

    /**
     * 落盘行程数据
     */
    static void saveTripData(SparkSession spark, JavaSparkContext jsc, List<SaveTripStatisticService> saveTripStatisticServices, TripStatisticExtendService tripStatisticExtendService, long d, JavaPairRDD<Long, AlgorithmOuterData> pairTidOutData) {
        JavaPairRDD<Long, List<OuterStatisticData>> tripRDD = pairTidOutData.mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, AlgorithmOuterData>>, Long, List<OuterStatisticData>>) tuple2Iterator ->
                new Iterator<Tuple2<Long, List<OuterStatisticData>>>() {
                    @Override
                    public boolean hasNext() {return tuple2Iterator.hasNext(); }
                    @Override
                    public Tuple2<Long, List<OuterStatisticData>> next() {
                        Tuple2<Long, AlgorithmOuterData> t = tuple2Iterator.next();
                        return new Tuple2<>(t._1(), t._2().getOutStatistics());
                    }
                });

        for (SaveTripStatisticService service : saveTripStatisticServices) {
            service.save(spark, jsc, tripRDD, d);
        }

        if(null != tripStatisticExtendService){
            tripStatisticExtendService.save(spark, jsc, tripRDD, d);
        }

        //手动清理资源
        tripRDD.unpersist();
    }

    /**
     * 落盘事件数据
     */
    static void saveEventData(SparkSession spark, JavaSparkContext jsc, List<SaveTripEventService> saveTripEventServices, long d, JavaPairRDD<Long, AlgorithmOuterData> pairTidOutData) {
        JavaPairRDD<Long, List<OuterEventData>> eventRDD = pairTidOutData.mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, AlgorithmOuterData>>, Long, List<OuterEventData>>) tuple2Iterator ->
                new Iterator<Tuple2<Long, List<OuterEventData>>>() {
                    @Override
                    public boolean hasNext() {
                        return tuple2Iterator.hasNext();
                    }
                    @Override
                    public Tuple2<Long, List<OuterEventData>> next() {
                        Tuple2<Long, AlgorithmOuterData> t = tuple2Iterator.next();
                        return new Tuple2<>(t._1(), t._2().getOutEvents());
                    }
                });

        for (SaveTripEventService service : saveTripEventServices) {
            service.save(spark, jsc, eventRDD, d);
        }

        //手动清理资源
        eventRDD.unpersist();
    }

    /**
     * 落盘行程发动机数据
     */
    static void saveTripEngineData(SparkSession spark, JavaSparkContext jsc, List<SaveTripEngineDataService> saveTripEngineDataServices, long d, JavaPairRDD<Long, AlgorithmOuterData> pairTidOutData) {
        //保存行程
        JavaPairRDD<Long, List<EngineAggregatedData>> tripRDD = pairTidOutData.mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, AlgorithmOuterData>>, Long, List<EngineAggregatedData>>) tuple2Iterator ->
                new Iterator<Tuple2<Long, List<EngineAggregatedData>>>() {
                    @Override
                    public boolean hasNext() {return tuple2Iterator.hasNext(); }
                    @Override
                    public Tuple2<Long, List<EngineAggregatedData>> next() {
                        Tuple2<Long, AlgorithmOuterData> t = tuple2Iterator.next();
                        return new Tuple2<>(t._1(), t._2().getOutEngineStatistics());
                    }
                });

        for (SaveTripEngineDataService service : saveTripEngineDataServices) {
            service.save(spark, jsc, tripRDD, d);
        }

        //手动清理资源
        tripRDD.unpersist();
    }

    /**
     * 落盘中间计算结果
     * @param spark
     * @param jsc
     * @param service
     * @param currentDay
     * @param outData
     * @param preDrivingData  昨天的中间数据
     */
    static void saveDrivingData(SparkSession spark, JavaSparkContext jsc, DrivingDataService service, long currentDay, JavaPairRDD<Long, AlgorithmOuterData> outData, final Broadcast<Map<Long, DrivingData>> preDrivingData) {
        JavaPairRDD<Long, DrivingData> drivingData = outData.mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, AlgorithmOuterData>>, Long, DrivingData>) iterator ->{
            List<Tuple2<Long, DrivingData>> result = new ArrayList(10000);
            while (iterator.hasNext()){
                Tuple2<Long, AlgorithmOuterData> next = iterator.next();
                java.util.Optional.ofNullable(next._2().getDrivingData()).ifPresent(e -> result.add(new Tuple2<>(next._1(), e)));
            }
            return result.iterator();
        });

        //合并昨天的中间数据（今天没有数据上传）
        Set<Long> nowTidSet = new HashSet<>(drivingData.keys().collect());
        List<Tuple2<Long, DrivingData>> yesterdayList = new ArrayList<>();
        for (Map.Entry<Long, DrivingData> entry : preDrivingData.getValue().entrySet()) {
            Long key = entry.getKey();
            if (!nowTidSet.contains(key)) {
                yesterdayList.add(new Tuple2<>(key, entry.getValue()));
            }
        }
        if (yesterdayList.size() > 0) {
            drivingData = drivingData.union(jsc.<Long,DrivingData>parallelizePairs(yesterdayList));
        }

        service.save(spark, jsc, drivingData, currentDay);
    }

}
